IICS(CDI)でSnowflakeからRedshiftにデータを移してみる
はじめに
データアナリティクス事業本部のkobayashiです。
Informatica Intelligent Cloud Services（以降、IICS）はオンプレミスおよびクラウド環境に置かれているデータサービスとアプリケーションサービスを統合するiPaaS（integration Platform as a Service）製品です。
IICSの機能の一つであるInformatica Cloud Data Integration（以降、CDI）ではクラウドやオンプレミスに存在する大量のデータをシームレスに統合します。 CDIではインフォマティカで用意してあるドライバを使うことで様々なデータソースから様々なターゲットにデータを統合することができます。
今回はSnowflakeのデータをRedshiftに移行してみたいと思います。
- 以前のCDIに関する記事
データ移行の内容と下準備
実施する内容としてはSnowflakeにはSF_IRISテーブルがありこれをRedshiftのsf_irisへ移行します。
図にすると以下の様になります。
前回は簡単にデータ移行を行うためCDIの実行環境としてInformatica Cloud Hosted Agentを使用しましたが、今回はAWSでEC2上にSecure Agentをインストールしてそこで実際のデータ移行を行うようにしたいと思います。Secure Agentのインストール・設定は別のエントリで詳しく解説されているのでその部分は割愛します。
CDIでソースやターゲットにするためには接続を作成する必要があります。すでに以前の記事でRedshiftの接続は作成してあるので今回はSnowflakeの接続のみを作成します。
Snowflakeの接続を作成
メニューから管理者機能へ進み接続設定を行う。
手順1）アドオンコネクタで無料トライアルを開始するを押下しコネクタを有効にする。
手順2）接続＞新しい接続を押下し、接続設定画面を開く。
手順3）下記のように接続設定を行い右上の接続テストを押下し接続のテストに成功しましたが表示されたら保存を押下し接続を作成する。
タイプ:Snowflake Cloud Data Warehouse V2を選択ランタイム接続: 事前に設定したSecure Agentを選択Username: Snowflakeへアクセスするユーザー名を入力Password: 上記のユーザのパスワードを入力Account:https://xxxxxxxxxx.snowflakecomputing.comのxxxxxxxxxxの部分を入力Warehouse: 使用したいWarehouse名を入力
これで接続の設定は終わったのでCDIでデータ移行の設定を行います。
SnowflakeからRedshiftへのデータ移行
今回行う手順は2つのステップです。
マッピングを作成するマッピングタスクを作成しタスクを実行してデータ移行を行う
メニューからデータ統合へ進み接続設定を行います。
マッピングの作成
手順1）左のメニューの新規を選択し、モーダルからマッピングを選択する。
手順2）マッピン、ソース、ターゲットそれぞれの全般タブの名前で適切な名前を付ける。
手順3）ソースステップを選択し、ソースタブを開いてデータソースの設定を行う。
- 接続 : 先に作成したSnowflakeの接続を選択
- オブジェクト :
選択を押下しデータソースを選択
手順4）処理日時を追加するためにソースとターゲットの間に式ステップを左のメニューからドラッグ＆ドロップする。
手順5）式ステップの式タブを選択しフィルドを追加する。
| フィールド | 式 |
|---|---|
| updatetime | Systimestamp()を入力 |
手順6）ターゲットステップを選択し、ターゲットタブを開いてデータソースの設定を行う。
- 接続 : 先に作成したRedshiftの接続を選択
- オブジェクト :
選択を押下しデータソースを選択
手順7）詳細を押下すると詳細設定が開くのでS3 Bucket NameでS3のバケット名を入力する。
手順8）フィールドマッピングタブを選択し、右下のオートマッチ＞正確なフィールド名を押下してターゲットのフィールドにマッピングを行う。
ここまででマッピングの設定は完成しました。
次にデータ移行を行うジョブを実行するマッピングタスクを作成します。
マッピングタスクの作成
手順1）左のメニューの新規を選択し、モーダルからマッピングタスクを選択する。
手順2）マッピングタスクの定義を設定し次へを謳歌する。
- タスク名 : 適切なタスク名を入力
- マッピング : 先に作成した
マッピングを選択
手順3）マッピングタスクのスケジュールを設定し完了を押下してマッピングタスクを作成する。
＊マッピングのジョブを定期実行する場合はスケジュールを設定しますが、今回は手動実行するので設定を行いません。
これでマッピングタスクが作成されました。
データ移行を実行する準備が整ったので実行してみます。
データ移行を実行（エラー発生）
手順1）作成したマッピングタスクを選択し右上の実行を押下する。
左のメニューからマイジョブを選択するとジョブの実行状況がわかりますが、この状態で実行した所エラーが発生してしまいました。
Snowflakeドライバでのエラーの解消
ジョブ実行時に表示されたエラーメッセージは以下になります。
[ERROR] Oct 20, 2020 3:45:57 PM net.snowflake.client.core.SFStatement executeQueryInternal SEVERE: Exception creating result java.lang.ExceptionInInitializerError at sun.misc.Unsafe.ensureClassInitialized(Native Method) at sun.reflect.UnsafeFieldAccessorFactory.newFieldAccessor(UnsafeFieldAccessorFactory.java:43) at sun.reflect.ReflectionFactory.newFieldAccessor(ReflectionFactory.java:156) at java.lang.reflect.Field.acquireFieldAccessor(Field.java:1088) at java.lang.reflect.Field.getFieldAccessor(Field.java:1069) at java.lang.reflect.Field.get(Field.java:393) at net.snowflake.client.jd
Javaの問題だとわかったのでこの解決方法を調べると全く同じ症状がInformaticaのナレッジに有りました。原因はJVMのメモリーの問題で解決方法もここに記載されていたのでその方法を実行します。
手順1）トップメニューで管理者を選択し、左のメニューからランタイム環境を選択する。その後、ジョブ実行しているSecure Agent名を押下する。
手順2）右上の編集を押下する。
手順3）JVMのメモリ設定を行い、右上の保存を押下する。
- サービス :
Data Integration Serverを選択 - タイプ :
DTMを選択 - DTM > JVMOption1 :
'-Xmx2048m'を入力
JVMのオプションを変更したため詳細画面に戻りエージェントの再起動を行う必要があります。
手順4）Secure Agentの詳細画面に戻り、サービスでData Integration Serverを選択し右の停止を押下する。
手順5）エージェントサービスの詳細でData Integration Serverが停止したことを確認した上でサービスでData Integration Serverを選択し右の開始を押下する。
これでジョブ実行を行うJVMのメモリエラーは解消されました。
データ移行を実行（JVM設定変更後）
先ほどと同じ手順でジョブを実行します。すると先ほどとは違いジョブが無事成功します。
移行先のRedshiftへもデータが移行されていることが確認できます。
まとめ
CDIを使うことで異なるサービス間でもGUI上で簡単にデータ移行を行えました。今回はSnowflakeからRedshiftへデータ移行を試してみました。この逆を行いたい場合もソースとターゲットを逆にするだけなのでそれほど手間なくデータ移行が行なえます。
CDIの良い点として接続さえきちんと作成できれば後はGUI上でデータの変換ができる点とSecure Agentを使用することでクローズドな環境でデータ操作を行える点かと思います。
最後まで読んで頂いてありがとうございました。
